package com.sensei.search.nodes;

import java.net.InetAddress;

import org.apache.log4j.Logger;

import com.linkedin.nimbus.NimbusException;
import com.linkedin.nimbus.cluster.Cluster;
import com.linkedin.nimbus.cluster.ClusterListenerAdapter;
import com.linkedin.nimbus.cluster.DefaultCluster;
import com.linkedin.nimbus.cluster.Node;
import com.linkedin.nimbus.cluster.NodeStatus;

public class SenseiNode extends ClusterListenerAdapter {
	private static Logger logger = Logger.getLogger(SenseiNode.class);
	
	private final String _zookeeperURL;
	private final int _id;
	private Cluster _cluster;
	private final SenseiServer _server;
	private final int[] _partitions;
	
	public SenseiNode(int id,int[] partitions,SenseiServer server,String zookeeperURL){
		_id = id;
		_partitions = partitions;
		_server = server;
		_zookeeperURL = zookeeperURL;
		_cluster = null;
	}

	@Override
	public void handleClusterSessionExpired(){
		logger.warn("Cluster was shut down, reconnecting...");
		try{
		  while(true){
			  try{
			    _cluster.awaitConnection();
			    break;
			  }
			  catch(InterruptedException ie){
				  continue;
			  }
		  }
		  
		  _cluster.available(_id);
		  logger.info("Reconnected to cluster");
		}
		catch(NimbusException ne){
			logger.error(ne.getMessage(),ne);
		}

	}

	public void startup() throws Exception{
		_server.startup();
		logger.info("connecting to cluster...");
		//_cluster = new DefaultCluster(_zookeeperURL,null,new DefaultClusterBootstrap(new RouterFactory(0)));
		_cluster = new DefaultCluster(_zookeeperURL,this);
		_cluster.awaitConnection();
		Node node = new Node(_id,InetAddress.getLocalHost(),_server.getPort(),_partitions,NodeStatus.UNAVAILABLE());
		try{
			_cluster.join(node);
		}
		catch(Exception e){
			// node prolly exists
			logger.warn(e.getMessage());
		}
		_cluster.available(_id);
		logger.info("started...");
	}
	
	public void shutdown() throws Exception{
		logger.info("shutting down...");
		try
		{
		  _cluster.leave(_id);
		}
		catch(Exception e){
			logger.warn(e.getMessage());
		}
		finally{
		  try{
		    _cluster.shutdown();
		    _cluster = null;
		  }
		  finally{
			_server.shutdown();
		  }
		}
	}
}
